package org.redis.cluster.statistics;

import java.util.concurrent.atomic.AtomicInteger;

import org.apache.commons.lang3.ArrayUtils;
import org.redis.cluster.event.Event;
import org.redis.cluster.event.EventListener;
import org.redis.cluster.interceptor.Interceptor;
import org.redis.cluster.interceptor.Invocation;
import org.redis.cluster.redis.RedisCommand;
import org.redis.cluster.redis.transport.RedisReply;
import org.redis.cluster.redis.transport.reply.RedisByteReply;
import org.redis.cluster.scheduler.second.SecondScheduleEvent;

public class TransportStatistics implements Interceptor,EventListener<SecondScheduleEvent> {

	private final static AtomicInteger maxInput=new AtomicInteger();
	private final static AtomicInteger maxOutput=new AtomicInteger();
	
	private final static AtomicInteger outputInSecond=new AtomicInteger();
	private final static AtomicInteger intputInSecond=new AtomicInteger();

	@Override
	public RedisReply intercept(Invocation invocation) throws Throwable {
		RedisCommand command=invocation.getCommand();
		if(ArrayUtils.isNotEmpty(command.getSource())) {
			int length=command.getSource().length;
			intputInSecond.addAndGet(length);
			updateMax(intputInSecond,length);
		}
		
		RedisReply reply=invocation.invoke();
		
		if(reply!=null && (reply instanceof RedisByteReply)) {
			RedisByteReply byteReply=(RedisByteReply)reply;
			if(ArrayUtils.isNotEmpty(byteReply.getValue())) {
				int length=byteReply.getValue().length;
				outputInSecond.addAndGet(length);
				updateMax(maxOutput,length);
			}
		}
		
		return reply;
	}
	
	@Override
	public void onEvent(SecondScheduleEvent event) {
		outputInSecond.set(0);
		intputInSecond.set(0);
	}
	
	@Override
	public boolean isSupport(Event event) {
		return (event instanceof SecondScheduleEvent);
	}
	
	protected boolean updateMax(AtomicInteger value,int newValue) {
		int oldValue=value.get();
		if(!(newValue>oldValue)) {
			return false;
		}
		
		return value.compareAndSet(oldValue, newValue);
	}
	
	public static int getMaxOutput() {
		return maxOutput.get();
	}
	
	public static int getMaxInput() {
		return maxInput.get();
	}
	
	public static int getOutputInSecond() {
		return outputInSecond.intValue();
	}
	
	public static int getInputInSecond() {
		return intputInSecond.intValue();
	}

}